using System;
using System.IO;
using System.Collections.Generic;
using System.Text;
using System.Diagnostics;
using System.Threading;
using System.Net;
using System.Net.Sockets;

namespace Microsoft.Samples.BizTalk.Adapter.Tcp.Receive
{
    /// <summary>
    ///     This class represents a main listener.
    /// </summary>
    /// <author>
    ///     Paul Brown
    /// </author>
    public class Listener
    {
        #region Private Instance Members

        private ReceiveAdapter  _receiveAdapter;    // The receive adapter
        private ReceiveLocation _receiveLocation;   // The receive location that is handling this duplex listener
        private Socket          _socket;            // The socket associated with the client connection
        private Thread          _listener;          // The listener thread used to start a socket and accept incoming connections
        private AutoResetEvent  _waitObject;        // This sync object allows us to wait until the duplex thread has finished terminating
        private bool            _terminateCalled;   // The adapter is being terminated by BizTalk
        private Dictionary<string, DuplexListener> _duplexListeners; // Represents multiple duplex listeners if this listener supports duplex connections
        private object          _lockObject;        // Lock object for duplex listeners collection
        private OrderedBatchWorkerThread _orderedWorkerThread; // The worker thread used for ordered batches

        // Delegate used to call Stop() method asynchronously on duplex listener
        private delegate void StopMethod();

        #endregion

        #region Constructor

        /// <summary>
        ///     This constructor initializes the class.
        /// </summary>
        public Listener(ReceiveAdapter adapter, ReceiveLocation receiveLocation)
        {
            // Validate arguments
            if (adapter == null) throw new ArgumentNullException("adapter");
            if (receiveLocation == null) throw new ArgumentNullException("receiveLocation");

            // Initialize members
            this._terminateCalled = false;
            this._receiveAdapter = adapter;
            this._receiveLocation = receiveLocation;
            this._socket = null;
            this._listener = null;
            this._waitObject = new AutoResetEvent(false);
            this._duplexListeners = new Dictionary<string, DuplexListener>();
            this._lockObject = new object();
            this._orderedWorkerThread = null;
        }

        #endregion

        #region Public Instance Events

        /// <summary>
        ///     This event is fired just before the listener thread terminates.
        /// </summary>
        public event EventHandler ThreadTerminated;

        #endregion

        #region Public Instance Properties

        /// <summary>
        ///     This property returns a reference to the receive adapter.
        /// </summary>
        public ReceiveAdapter Adapter
        {
            get { return (this._receiveAdapter); }
        }

        /// <summary>
        ///     This property returns a reference to the receive location.
        /// </summary>
        public ReceiveLocation ReceiveLocation
        {
            get { return (this._receiveLocation); }
        }

        #endregion

        #region Public Instance Methods

        /// <summary>
        ///     This method starts the main listener.
        /// </summary>
        public void Start()
        {
            if (this.Adapter.TraceSwitch.TraceVerbose)
                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveStartingMainListener, this.ReceiveLocation.Uri);

            // Set up an ordered worker thread if this receive location is to be ordered
            if (this.ReceiveLocation.IsOrdered)
            {
                if (this.Adapter.TraceSwitch.TraceVerbose)
                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveStartingOrderedWorkerThread, this.ReceiveLocation.Uri);

                // Start thread
                this._orderedWorkerThread = new OrderedBatchWorkerThread(this.Adapter);
                this._orderedWorkerThread.Start();
            }

            // Create listener
            if (this.Adapter.UseBlockingSockets)
            {
                this.CreateSyncListener();
            }
            else
            {
                this.CreateAsyncListener();
            }

            if (this.Adapter.TraceSwitch.TraceVerbose)
                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveStartedMainListener, this.ReceiveLocation.Uri);
        }

        /// <summary>
        ///     This method stops the main listener and all duplex listeners.  It will block until it has stopped.
        /// </summary>
        public void Stop()
        {
            if (this.Adapter.TraceSwitch.TraceVerbose)
                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveStoppingMainListener, this.ReceiveLocation.Uri);

            // Set flag
            this._terminateCalled = true;

            // Stop duplex listeners
            if (this.ReceiveLocation.IsDuplex)
            {
                lock (this._lockObject)
                {
                    foreach (KeyValuePair<string, DuplexListener> kvp in this._duplexListeners)
                    {
                        // Unsubscribe from event
                        kvp.Value.ThreadTerminated -= new EventHandler(this.DuplexListener_ThreadTerminated);

                        // Stop listener
                        kvp.Value.Stop();
                    }
                }
            }

            // Close the main socket to force Socket.Accept() to exit
            if (this._socket != null)
            {
                this._socket.Close(this.ReceiveLocation.MainCloseTimeout);
            }

            // Only wait if we have a blocking thread (as it will have been sitting in Socket.Accept())
            if (this.Adapter.UseBlockingSockets)
            {
                // Wait for listener to stop
                int timeout = (this.ReceiveLocation.ListenerTerminateTimeout == -1 ? Timeout.Infinite : this.ReceiveLocation.ListenerTerminateTimeout);
                bool receivedSignal = this._waitObject.WaitOne(timeout, true);

                if (!receivedSignal)
                {
                    // Reset flag so that the wait object isn't signalled in listener thread
                    this._terminateCalled = false;

                    // We timed out, abort threads!
                    if (this.Adapter.UseBlockingSockets)
                    {
                        if (this._listener.ThreadState != System.Threading.ThreadState.Stopped &&
                            this._listener.ThreadState != System.Threading.ThreadState.Aborted)
                        {
                            this._listener.Abort();
                        }
                    }
                }
            }

            // Stop ordered worker thread if we are ordered
            if (this.ReceiveLocation.IsOrdered)
            {
                if (this.Adapter.TraceSwitch.TraceVerbose)
                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveStoppingOrderedWorkerThread, this.ReceiveLocation.Uri);

                // Stop thread
                this._orderedWorkerThread.Stop();
            }

            if (this.Adapter.TraceSwitch.TraceVerbose)
                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveStoppedMainListener, this.ReceiveLocation.Uri);
        }

        #endregion

        #region Protected Instance Methods

        /// <summary>
        ///     This method invokes the ThreadTerminated event.
        /// </summary>
        protected virtual void OnThreadTerminated(EventArgs e)
        {
            if (this.ThreadTerminated != null)
            {
                this.ThreadTerminated(this, e);
            }
        }

        #endregion

        #region Private Instance Methods

        /// <summary>
        ///     This method starts a dedicated listening thread for the receive location.
        /// </summary>
        private void CreateSyncListener()
        {
            if (this.Adapter.TraceSwitch.TraceVerbose)
                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveCreatingSyncListener, this.ReceiveLocation.ListenPort);

            // Create and start thread
            this._listener = new Thread(new ThreadStart(this.Listen));
            this._listener.Name = TraceHelper.ResourceManager.GetString(ResourceConstants.ListenerThreadNamePrefix) + this._listener.ManagedThreadId.ToString();
            this._listener.IsBackground = true;
            this._listener.Start();
        }

        /// <summary>
        ///     This method starts an asynchronous listening thread that uses the .NET thread pool for the receive location.
        /// </summary>
        private void CreateAsyncListener()
        {
            if (this.Adapter.TraceSwitch.TraceVerbose)
                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveCreatingAsyncListener, this.ReceiveLocation.ListenPort);

            // Create socket, bind, listen
            this._socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            this._socket.Bind(new IPEndPoint(IPAddress.Any, this.ReceiveLocation.ListenPort));
            this._socket.Listen(this.ReceiveLocation.MaxPendingConnections);

            // Begin accepting connections
            IAsyncResult result = this._socket.BeginAccept(new AsyncCallback(this.AcceptCallback), null);

            if (this.Adapter.TraceSwitch.TraceVerbose)
                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveStartingMainAsyncListener);
        }

        /// <summary>
        ///     This method is a ThreadStart delegate whose purpose is to start up a TCP listener.
        /// </summary>
        private void Listen()
        {
            if (this.Adapter.TraceSwitch.TraceVerbose)
                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveStartingMainListenerSyncThread);

            bool socketClosed = false;
            try
            {
                // Create socket, bind and then listen
                this._socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
                this._socket.Bind(new IPEndPoint(IPAddress.Any, this.ReceiveLocation.ListenPort));
                this._socket.Listen(this.ReceiveLocation.MaxPendingConnections);

                // Block in thread until it is time to stop
                while (!this._terminateCalled)
                {
                    // Client socket variable
                    Socket clientSocket = null;

                    if (this.Adapter.TraceSwitch.TraceVerbose)
                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveBlockingOnAccept, this.ReceiveLocation.ListenPort);

                    // Sit in Accept() waiting for new connections
                    clientSocket = this._socket.Accept();

                    if (this.Adapter.TraceSwitch.TraceVerbose)
                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveAcceptedNewSocket, this.ReceiveLocation.ListenPort, (clientSocket.RemoteEndPoint as IPEndPoint).Address, (clientSocket.RemoteEndPoint as IPEndPoint).Port);

                    // Use socket option KeepAlive to ensure that every 2 hours the server sends a KeepAlive message to see if the
                    // client still exists and to stop it timing out.
                    clientSocket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, 1);

                    // Set socket options: receive timeout, send timeout, receive buffer size, send buffer size
                    clientSocket.ReceiveBufferSize = this.ReceiveLocation.SocketReceiveBufferSize;
                    clientSocket.ReceiveTimeout = this.ReceiveLocation.SocketReceiveTimeout;
                    clientSocket.SendBufferSize = this.ReceiveLocation.SocketSendBufferSize;
                    clientSocket.SendTimeout = this.ReceiveLocation.SocketSendTimeout;

                    if (this.ReceiveLocation.IsDuplex)
                    {
                        // Create duplex listener
                        DuplexListener duplexListener = new DuplexListener(this.Adapter, this.ReceiveLocation, clientSocket);
                        duplexListener.ThreadTerminated += new EventHandler(this.DuplexListener_ThreadTerminated);

                        // Start it
                        duplexListener.Start();
                        lock (this._lockObject)
                        {
                            // Add new duplex listener to collection
                            this._duplexListeners.Add(duplexListener.DuplexUrl, duplexListener);
                        }
                    }
                    else
                    {
                        // Pass socket to dispatcher only if this is not an ordered receive location, otherwise deal with
                        // socket by passing work to an OrderedBatchWorkerThread directly and waiting.
                        if (this.ReceiveLocation.IsOrdered)
                        {
                            if (this.Adapter.TraceSwitch.TraceVerbose)
                                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveSubmittingOrderedWorkItem, this.ReceiveLocation.Uri);

                            // Pass socket to ordered worker thread directly (this will block until it is finished)
                            this._orderedWorkerThread.DoWork(new OrderedBatchWorkItem(this.ReceiveLocation, clientSocket) as IWorkItem);
                        }
                        else
                        {
                            // Pass socket to dispatcher
                            this.Adapter.Dispatcher.AddWorkItem(new BatchWorkItem(this.ReceiveLocation, clientSocket) as IWorkItem);
                        }
                    }
                }

                if (this.Adapter.TraceSwitch.TraceVerbose)
                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveTerminateCalled, Thread.CurrentThread.Name);
            }
            catch (ObjectDisposedException e)
            {
                if (this.Adapter.TraceSwitch.TraceVerbose)
                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveSocketDisposedError, e.Message);

                // Socket has been closed by Stop(), just end
                socketClosed = true;
            }
            catch (ThreadAbortException e)
            {
                // We have been aborted by a call to Thread.Abort()
                TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.ReceiveThreadAbortedError, e.Message);

                // Socket has been closed by Stop(), just end
                socketClosed = true;
            }
            catch (SocketException e)
            {
                // Handle socket error
                SocketErrorHandler.HandleError(this.Adapter, this.ReceiveLocation, e, true);
            }
            catch (Exception e)
            {
                TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.ReceiveGeneralError, e.Message);

                // We have aborted somehow but not by Stop(), close down receive location
                this.Adapter.TransportProxy.ReceiverShuttingdown(this.ReceiveLocation.Uri, e);
            }
            finally
            {
                // Close the socket
                if (!socketClosed)
                {
                    this._socket.Close(this.ReceiveLocation.MainCloseTimeout);
                }

                // Terminate the thread cleanly if we can
                if (this._terminateCalled)
                {
                    // We have been terminated so release Stop() method
                    this._waitObject.Set();
                }

                // Fire terminate event
                this.OnThreadTerminated(new EventArgs());
            }
        }

        /// <summary>
        ///     This method represents the delegate for the asynchronous callback of the BeginAccept() call on the socket.
        /// </summary>
        private void AcceptCallback(IAsyncResult asyncResult)
        {
            if (this.Adapter.TraceSwitch.TraceVerbose)
                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveSocketAcceptCallbackInvoked);

            try
            {
                // Accept new socket
                Socket clientSocket = this._socket.EndAccept(asyncResult);

                if (this.Adapter.TraceSwitch.TraceVerbose)
                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveAcceptedNewSocket, this.ReceiveLocation.ListenPort, (clientSocket.RemoteEndPoint as IPEndPoint).Address, (clientSocket.RemoteEndPoint as IPEndPoint).Port);

                // Use socket option KeepAlive to ensure that every 2 hours the server sends a KeepAlive message to see if the
                // client still exists and to stop it timing out.
                clientSocket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, 1);

                // Set socket options: receive timeout, send timeout, receive buffer size, send buffer size
                clientSocket.ReceiveBufferSize = this.ReceiveLocation.SocketReceiveBufferSize;
                clientSocket.ReceiveTimeout = this.ReceiveLocation.SocketReceiveTimeout;
                clientSocket.SendBufferSize = this.ReceiveLocation.SocketSendBufferSize;
                clientSocket.SendTimeout = this.ReceiveLocation.SocketSendTimeout;

                // If we are a duplex enabled receive location
                if (this.ReceiveLocation.IsDuplex)
                {
                    // Create duplex listener
                    DuplexListener duplexListener = new DuplexListener(this.Adapter, this.ReceiveLocation, clientSocket);

                    // Start it
                    duplexListener.Start();
                    lock (this._lockObject)
                    {
                        // Add new duplex listener to collection
                        this._duplexListeners.Add(duplexListener.DuplexUrl, duplexListener);
                    }
                }
                else
                {
                    // Pass socket to dispatcher only if this is not an ordered receive location, otherwise deal with
                    // socket by passing work to an OrderedBatchWorkerThread directly and waiting.
                    if (this.ReceiveLocation.IsOrdered)
                    {
                        if (this.Adapter.TraceSwitch.TraceVerbose)
                            TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveSubmittingOrderedWorkItem, this.ReceiveLocation.Uri);

                        // Pass socket to ordered worker thread directly (this will block until it is finished)
                        this._orderedWorkerThread.DoWork(new OrderedBatchWorkItem(this.ReceiveLocation, clientSocket) as IWorkItem);
                    }
                    else
                    {
                        // Pass socket to dispatcher
                        this.Adapter.Dispatcher.AddWorkItem(new BatchWorkItem(this.ReceiveLocation, clientSocket) as IWorkItem);
                    }
                }
            }
            catch (ObjectDisposedException e)
            {
                if (this.Adapter.TraceSwitch.TraceVerbose)
                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveSocketDisposedError, e.Message);

                // Socket closed - shut down
                this._terminateCalled = true;
            }
            catch (SocketException e)
            {
                // Handle socket error
                SocketErrorHandler.HandleError(this.Adapter, this.ReceiveLocation, e, true);
            }
            catch (Exception e)
            {
                TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.ReceiveGeneralError, e.Message);

                // We have aborted somehow but not by Stop(), close down receive location
                this.Adapter.TransportProxy.ReceiverShuttingdown(this.ReceiveLocation.Uri, e);
            }
            finally
            {
                // Accept new sockets if we are not terminating
                if ((!this._terminateCalled) && (this._socket != null))
                {
                    IAsyncResult result = this._socket.BeginAccept(this.AcceptCallback, null);
                }
            }
        }

        #endregion

        #region Private Instance Event Handlers

        /// <summary>
        ///     This event handler is called whenever the duplex listener thread has terminated unexpectedly without
        ///     first calling Stop() in this class.
        /// </summary>
        private void DuplexListener_ThreadTerminated(object sender, EventArgs e)
        {
            // Invoke Stop() on duplex object asynchronously to avoid running on same thread as the event invocation
            DuplexListener listener = (DuplexListener)sender;
            StopMethod stopMethod = new StopMethod(listener.Stop);

            // Invoke method
            IAsyncResult result = stopMethod.BeginInvoke(null, null);

            // Wait for it to finish
            result.AsyncWaitHandle.WaitOne();

            // Remove from array
            lock (this._lockObject)
            {
                this._duplexListeners.Remove(listener.DuplexUrl);
            }
        }

        #endregion
    }
}
